package com.atguigu.sparkstreaming.demos

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}

/**
 * Created by Smexy on 2022/5/23
 */
object SparkStreamingKafkaDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "SparkStreamingKafkaDemo", Seconds(5))

    /*
        定义消费者的参数。
          必须传入:
            bootstrap.servers : kafka集群地址。
                      不建议写太多，消费者在启动时，会依次向写的集群地址的每一个server都尝试建议一个tcp连接，
                      返回最早成功的那个。
            key,value的反序列化器: key.deserializer
                                value.deserializer
           group.id: 消费者组id
           auto.offset.reset： 从哪个位置开始消费。
                    latest : 从当前主题的最新位置往后消费
                    earliest:  如果当前主题没有被这个组消费过，从主题最早的位置消费
            enable.auto.commit： 是否允许自动提交offset(提交到kafka内部的_consumer_offsets主题中)

     */
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "atguigu1227",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    /*
        消费的主题。
            通常一个 DStream 只消费一个主题。
            如果消费多个主题，一般是创建多个DStream
     */
    val topics1 = Array("topicA")

    /*
        使用kafka提供的工具类，创建DStream(流，代表源源不断地去消费主题中的数据，所组成的无限集合)
     */
    val stream = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      LocationStrategies.PreferConsistent,
      ConsumerStrategies.Subscribe[String, String](topics1, kafkaParams)
    )


    // 转换运算
    val ds: DStream[String] = stream.map(record => record.value)

    // 输出结果
    ds.print(1000)

    // 启动app
    streamingContext.start()

    // 阻塞当前线程，让程序24h不停运行
    streamingContext.awaitTermination()


  }

}
